package com.navinfo.tripanalysis.offline.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.navinfo.opentsp.platform.location.protocol.common.LCStatusType;
import com.navinfo.tripanalysis.common.pojo.Point0200;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.pojo.StatusAddition;
import com.navinfo.tripanalysis.offline.service.DataBatchService;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConverters;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

import static com.navinfo.opentsp.platform.location.protocol.common.LCStatusType.StatusType.*;

/**
 * 位置点的工具类
 */
public class PointUtils {
    private static final Logger logger = LoggerFactory.getLogger(PointUtils.class);
    /**
     * 获取0200\0F37\公共轨迹协议表按批次加载数据所对应的HDFS完整文件夹
     * @param basePath  hdfs文件根路径
     * @param param 分批加载数据对应的参数
     * @return
     */
    public static List<String> getHDFSPath(String basePath, LoadDataParam param) {
        if(param == null){
            logger.error("获取按批加载数据所对应的HDFS完整文件夹时，LoadDataParam为null....");
            return null;
        }

        Date day =  param.getDay();
        List<Long> tidList = param.getTidList();
        List<Integer> batchList = param.getBatchList();

        List<String> list = new ArrayList<>();
        String dayStr = DateUtils.format(day, DateUtils.DateFormat.YYYYMMDD);
        try {
            FileSystem fs = FileSystem.get(new java.net.URI(basePath), new org.apache.hadoop.conf.Configuration());

            //是否按批次加载
            if (CommonUtils.isCollectionEmpty(batchList)) {
                //是否加载特定tid
                if(CommonUtils.isCollectionEmpty(tidList)){
                    String dataPathStr = String.format("%s/part_time=%s", basePath, dayStr);
                    if (fs.exists( new Path(dataPathStr))) {
                        list.add(dataPathStr);
                    } else {
                        logger.info("dataPathStr:{} is not exist", dataPathStr);
                    }
                }else{
                    tidList.stream().map(tid -> tid.intValue() % DataBatchService.MAX).
                            collect(Collectors.toSet()).
                            forEach(hashTid -> {
                                String dataPathStr = String.format("%s/part_time=%s/hashtid=%d", basePath, dayStr, hashTid);
                                try {
                                    if (fs.exists( new Path(dataPathStr))) {
                                        list.add(dataPathStr);
                                    }else {
                                        logger.info("dataPathStr:{} is not exist", dataPathStr);
                                    }
                                } catch (IOException e) {
                                    logger.info("dataPathStr:{} is not exist", dataPathStr);
                                }
                            });
                }
            } else {
                for (Integer id : batchList) {
                    String dataPathStr = String.format("%s/part_time=%s/hashtid=%d", basePath, dayStr, id);
                    Path dataPath = new Path(dataPathStr);
                    if (fs.exists(dataPath)) {
                        list.add(dataPathStr);
                    } else {
                        logger.info("dataPathStr:{} is not exist", dataPathStr);
                    }
                }
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return list;
    }

    /**
     * 获取0200\0F37\公共轨迹协议表按批次加载数据所对应的Where条件
     * @param param  分批加载数据对应的参数
     * @return
     */
    public static String getBatchWhereSql(LoadDataParam param) {
        if(param == null){
            logger.error("获取按批加载数据的Where条件时，LoadDataParam为null....");
            return "";
        }

        Date day =  param.getDay();
        List<Long> tidList = param.getTidList();
        List<Integer> batchList = param.getBatchList();

        String dayStr = DateUtils.format(day, DateUtils.DateFormat.YYYYMMDD);
        String rtn = "";
        try {
            //是否分区
            if (CommonUtils.isCollectionEmpty(batchList)) {
                //是否加载特定tid
                if(CommonUtils.isCollectionEmpty(tidList)){
                    rtn = String.format(" part_time=%s ", dayStr);
                }else{
                    //Set<Integer> tidSet = tidList.stream().map(tid -> tid.intValue() % DataBatchService.MAX).collect(Collectors.toSet());
                    //测试数据tid超过int最大值,临时调整
                    Set<Integer> tidSet = tidList.stream().map(tid -> Integer.valueOf((int)(tid % DataBatchService.MAX))).collect(Collectors.toSet());
                    rtn = String.format(" part_time=%s and hashtid in (%s) ", dayStr,  CommonUtils.mkString(tidSet, ","));
                }
            } else {
                rtn = String.format(" part_time=%s and hashtid in (%s) ", dayStr, CommonUtils.mkString(batchList, ","));
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return rtn;
    }

    /**
     * 获取0F39\0F3D无hashtid分区，无法按批次加载数据所对应的Where条件
     * 对于没有hashtid分区的基础协议，where条件过滤掉hashtid分区加载条件，只按日期加载
     * @param param
     * @return
     */
    public static String getBatchWhereSqlWihtoutHashtid(LoadDataParam param){
        LoadDataParam paramWithoutHashtid = new LoadDataParam();
        paramWithoutHashtid.setDay(param.getDay());
        paramWithoutHashtid.setExtendParams(param.getExtendParams());
        return getBatchWhereSql(paramWithoutHashtid);
    }

    /**
     * 获取0F39\0F3D无hashtid分区，无法按批次加载数据所对应的HDFS完整文件夹
     * 对于没有hashtid分区的基础协议，where条件过滤掉hashtid分区加载条件，只按日期加载
     * @param param
     * @return
     */
    public static List<String> getHDFSPathWithoutHashtid(String basePath, LoadDataParam param){
        LoadDataParam paramWithoutHashtid = new LoadDataParam();
        paramWithoutHashtid.setDay(param.getDay());
        paramWithoutHashtid.setExtendParams(param.getExtendParams());
        return getHDFSPath(basePath,paramWithoutHashtid);
    }


    /**
     * 获取0f3d表按批次加载数据所对应的Where条件
     * @param param
     * @return
     */
    public static String getBatchWhereSql4yq(LoadDataParam param) {
        if(param == null){
            logger.error("获取按批加载数据的Where条件时，LoadDataParam为null....");
            return "";
        }

        Date day =  param.getDay();
        List<Long> tidList = param.getTidList();
        List<Integer> batchList = param.getBatchList();

        String dayStr = DateUtils.format(day, DateUtils.DateFormat.YYYYMMDD);
        String rtn = "";
        try {
            //是否分区
            if (CommonUtils.isCollectionEmpty(batchList)) {
                //是否加载特定tid
                if(CommonUtils.isCollectionEmpty(tidList)){
                    rtn = String.format(" part_time=%s ", dayStr);
                }else{
                    //Set<Integer> tidSet = tidList.stream().map(tid -> tid.intValue() % DataBatchService.MAX).collect(Collectors.toSet());
                    //测试数据tid超过int最大值,临时调整
                    Set<Integer> tidSet = tidList.stream().map(tid -> Integer.valueOf((int)(tid % DataBatchService.MAX))).collect(Collectors.toSet());
                    rtn = String.format(" part_time=%s", dayStr);
                }
            } else {
                rtn = String.format(" part_time=%s", dayStr);
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return rtn;
    }

    /**
     * 车辆状态附加位赋值到Point0200中
     * @param point
     * @param statusAddition
     */
    public static void setStatusAddition(Point0200 point, String statusAddition){
        if(point!=null && statusAddition!=null) {
            List<StatusAddition> statusAdditions = parseToList(statusAddition);
            praseStatusAddition(point,statusAdditions);
        }
    }

    /**
     * 车辆状态附加位赋值到Point0200中
     * @param point
     * @param statusAddition
     */
    public static void setStatusAddition4YQ(Point0200 point, Object statusAddition){
        if(point!=null && !org.springframework.util.StringUtils.isEmpty(statusAddition)) {
            List<StatusAddition> statusAdditions = parseToList4YQ(statusAddition);
            praseStatusAddition(point,statusAdditions);
        }
    }

    /**
     * 车辆附加报警位赋值到Point0200中
     * @param point
     * @param additionAlarm
     */
    public static void setAdditionAlarm(Point0200 point, String additionAlarm){
        if (StringUtils.startsWith(additionAlarm, "[")) {
            List<Byte> dataList = JSON.parseObject(additionAlarm,  new TypeReference<List<Byte>>() {});
            if (dataList.size() >= 8) {
                //0200特有，在第8个字节 Bit1 手刹开关
                point.setHandBrakeStatus((dataList.get(7) >> 1) & 0x01);
            }
            if (dataList.size() >= 6) {
                //在第6个字节
                byte additionAlarmAt6 = dataList.get(5);
                //bit0 空挡滑行，0200特有
                point.setSlideNeutralStatus(additionAlarmAt6 & 0x01);
                //bit1 急加速，0200特有
                point.setRapidAccelerationStatus((additionAlarmAt6 >> 1) & 0x01);
                //bit2 急减速，0200特有
                point.setRapidDecelerationStatus((additionAlarmAt6 >> 2) & 0x01);
                //bit3 急转弯，0200特有
                point.setSharpTurningStatus((additionAlarmAt6 >> 3) & 0x01);
                //bit6 猛踩油门报警状态，0200特有
                point.setSlamAcceleratorStatus((additionAlarmAt6 >> 6) & 0x01);
            }
        }
    }

    public static void setAdditionAlarm(Point0200 point,int eventId){
        if(eventId > 0){
            switch(eventId){
                //急减速事件
                case 10:
                    point.setRapidDecelerationStatus(1);break;
                //急加速事件
                case 12:
                    point.setRapidAccelerationStatus(1);break;
                //急转弯事件
                case 14:
                    point.setSharpTurningStatus(1);break;
                default:break;
            }
        }
    }

    private static List<StatusAddition> parseToList(String statusStr) {
        if (statusStr == null) {
            return Collections.emptyList();
        }
        TypeReference<List<String>> typeReference = new TypeReference<List<String>>() {};
        List<String> dataList = JSON.parseObject(statusStr, typeReference);
        List<StatusAddition> statusList = new ArrayList<>();
        for (String data : dataList) {
            String[] arr = data.split("_");
            statusList.add(new StatusAddition(arr[0], Long.parseLong(arr[1])));
        }
        return statusList;
    }


    private static List<StatusAddition> parseToList4YQ (Object obj){

        if(org.springframework.util.StringUtils.isEmpty(obj)){
            return Collections.emptyList();
        }
        List<StatusAddition> status = new ArrayList<>();
        //logger.info("statusAddition:{}",statusAddition);
        GenericRowWithSchema q = (GenericRowWithSchema)obj;
        //GenericRowWithSchema a = q.getAs("statusaddition");
        scala.collection.mutable.Seq<GenericRowWithSchema> b = q.getAs("status");
        //增加判断复杂类型内部是否为空
        if(b!= null) {
            List<GenericRowWithSchema> c = JavaConverters.mutableSeqAsJavaListConverter(b).asJava();
            for (GenericRowWithSchema r : c) {
                String name = String.valueOf(Optional.ofNullable(r.getAs("types")).orElse(""));
                String key = String.valueOf(LCStatusType.StatusType.valueOf(name).getNumber());
                long value = Long.parseLong(String.valueOf(Optional.ofNullable(r.getAs("statusvalue")).orElse("0")));
                StatusAddition e = new StatusAddition(key, value);
                status.add(e);
            }
        }
        return status;
    }

    /**
     * 将解析出来附加位信息保存到Point0200
     * @param point
     * @param statusAdditions
     */
    private static void praseStatusAddition (Point0200 point,List<StatusAddition> statusAdditions){
        for (StatusAddition status : statusAdditions) {
            long statusValue = status.getValue();
            switch (Integer.parseInt(status.getKey())) {
                case rotation_VALUE:
                    //当前转速
                    point.setRotation(new BigDecimal(statusValue).divide(new BigDecimal("100"), 0, BigDecimal.ROUND_HALF_UP).intValue());
                    break;
                case mileage_VALUE:
                    //0200特有，can里程(ecu)
                    point.setEcuMileage(statusValue / 100.0);
                    break;
                case mileageDD_VALUE:
                    point.setMeterMileage(new BigDecimal(statusValue).divide(new BigDecimal("100"),3,BigDecimal.ROUND_HALF_UP).doubleValue());//0200特有，仪表里程
                    break;
                case differentialMileage_VALUE:
                    point.setIntegralMileage(statusValue / 100.0);//0200特有，积分里程，不使用
                    break;
                case totalFuelConsumption_VALUE:
                    point.setTotalFuelCons(new BigDecimal(statusValue).divide(new BigDecimal("100"),3,BigDecimal.ROUND_HALF_UP).doubleValue());//0200特有，燃油总消耗
                    break;
                case integralFuelConsumption_VALUE:
                    point.setIntegralFuelCons(statusValue / 100.0);//0200特有，积分油耗
                    break;
                case accPedalPos_VALUE:
                    point.setAccPedalPos((int) statusValue / 100);//油门开度
                    break;
                case coolingWaterTem_VALUE:
                    point.setCoolingWaterTem((int) statusValue / 100);//0200特有，冷却水温
                    break;
                case intakeAirTem_VALUE:
                    point.setIntakeAirTem((int) statusValue / 100); //0200特有，进气温度
                    break;
                case dpfPressure_VALUE:
                    point.setIntakePressure((int) statusValue / 100); //0200特有，进气压力
                    break;
                case clutchSwitch_VALUE:
                    point.setClutchSwitch((int) statusValue / 100); //离合器开关
                    break;
                case cruiseCtrlActive_VALUE:
                    point.setCruiseControl((int) statusValue); //0200特有，定速巡航开关
                    break;
                case oilValue_VALUE:
                    //0200特有，油液百分比
                    double per = statusValue / 100.0;
                    if (per >= 0 && per <= 100) {
                        point.setOilPercentage(per);
                    }
                    break;
                case fuelConsumptionRate_VALUE:
                    point.setFuelConsumptionRate(statusValue / 100.0 / 1000.0); //发动机燃油消耗率（毫升/小时）转为 L/H
                    break;
                case realTimeOilConsumption_VALUE:
                    point.setRealTimeOilConsumption(statusValue / 100.0);//实时油耗(发动机瞬时油耗)（KM/L）
                    break;
                case actualEnginePercentTor_VALUE:
                    point.setEngineOutputTorque((int) statusValue / 100); //实际发动机转矩百分比
                    break;
                case cumulativeRunningTime_VALUE:
                    point.setCumulativeRunningTime(statusValue / 100); //累计运行时长
                    break;
                default:
                    break;
            }
        }
    }

}

